# Основы параллелизма Cats с Ref и Deferred

[Перевод статьи "Cats Concurrency basics with Ref and Deferred", автор Krzysztof Grajek][Ref and Deffered]

## Параллельный доступ и ссылочная прозрачность

`Ref` и `Deferred` являются основными строительными блоками структур в ФП, используемых параллельно. 
Особенно при использовании с tagless final абстракцией 
эти два элемента при построении бизнес-логики могут дать нам **и параллельный доступ, и ссылочную прозрачность**, 
и мы можем использовать их для создания более сложных структур, таких как счетчики и конечные автоматы.

Прежде чем углубиться в `Ref` и `Deferred`, полезно знать, 
что параллелизм в Cats основан на Java `AtomicReference`, и мы начнем наше путешествие здесь.

## Atomic Reference

`AtomicReference` является одним из элементов пакета `java.util.concurrent.atomic`, 
и [в документах Oracle](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/package-summary.html) 
мы можем прочитать, что `java.util.concurrent.atomic` это:

> Небольшой набор классов, поддерживающих безблокировочное поточно-ориентированное программирование с одиночными переменными. 
> По сути, классы в этом пакете расширяют понятие `volatile` значений, полей и элементов массива 
> до тех, которые также обеспечивают атомарную операцию условного обновления.
> 
> ...
> 
> Экземпляры классов [AtomicBoolean](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/AtomicBoolean.html), 
> [AtomicInteger](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/AtomicInteger.html), 
> [AtomicLong](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/AtomicLong.html)
> и [AtomicReference](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/AtomicReference.html)
> каждый обеспечивают доступ и обновления к одной переменной соответствующего типа.

`AtomicReference` с нами начиная с Java 1.5, и он используется для 
получения **большей производительности, чем синхронизация** (что обычно, но не всегда).

Когда вам нужно поделиться некоторыми данными между потоками, то вы должны защитить доступ к этой части данных. 
Простейшим примером может быть увеличение некоторого `int`: `i = i + 1`. 
Наш пример состоит на самом деле из 3 операций: 
сначала мы читаем значение `i`, затем прибавляем `1` к этому значению, 
а в конце присваиваем `i` вновь вычисленное значение. 
В многопоточных приложениях может возникнуть ситуация, 
когда каждый поток будет выполнять эти 3 шага между шагами других потоков, 
и окончательное значение `i` невозможно будет предсказать.

Обычно возникают мысли о механизме слова `synchronised` или класса блокировки, 
но с `atomic.*` больше не нужно беспокоиться о явной синхронизации, 
и вы можете использовать предоставленные служебные атомарные типы, 
где автоматически включается проверка, выполняется ли операция за один шаг.

Возьмем в качестве примера `AtomicInteger.incrementAndGet`:

```java
/**
 * Atomically increments by one the current value.
 *
 * @return the updated value
 */
public final int incrementAndGet() {
    for (;;) {
        int current = get();
        int next = current + 1;
        if (compareAndSet(current, next))
            return next;
    }
}
```

С помощью операции `compareAndSet` мы либо обновляем наши данные, либо терпим неудачу, 
но мы никогда не заставляем поток ждать. 
Таким образом, если `compareAndSet` в `incrementAndGet` не удается, 
то мы просто пытаемся повторить всю операцию снова, извлекая текущее значение данных в `get()`. 
С другой стороны, с `synchronized`-подобными механизмами нет ограничений на количество операторов, 
которые можно выполнить, пока блокировка получена, 
но этот блок никогда не выйдет из строя и может заставить вызывающий поток ждать, 
обеспечивая вероятность взаимоблокировки или снижения производительности.

Теперь, разобрав основы, давайте перейдем к нашей первой мегазвезде параллелизма.

## Ref

Cats `Ref` очень похож на упомянутую выше атомарную ссылку Java, 
основные отличия в том, что первое - `Ref` используется с tagless final абстракцией `F`, 
второе - оно всегда содержит значение, 
и значение, содержащееся в `Ref` — типа `A`, - это неизменяемое значение.

```dotty
abstract class Ref[F[_], A]:
  def get: F[A]
  def set(a: A): F[Unit]
  def modify[B](f: A => (A, B)): F[B]
  // ... и т.д.
```

`Ref[F[_], A]` является чисто функциональной изменяемой ссылкой:

- параллельной
- свободно блокируемой
- и всегда содержит значение

Он создается путем присвоения начального значения, и каждая операция оборачивается в `F`, например, в `cats.effect.IO`.

Если внимательно посмотреть на сопутствующий объект для Cats `Ref`, 
то увидим, что необходимо предоставить также `Sync[F]`:

```dotty
def of[F[_], A](a: A)(implicit F: Sync[F]): F[Ref[F, A]] = F.delay(unsafe(a))
```

Приведенный выше метод является лишь примером многих операций, доступных в `Ref`, 
он используется для построения `Ref` с начальным значением. 
`Sync` дает возможность приостановить любые побочные эффекты 
с помощью своего метода `delay` для каждой операции на `Ref`.

`Ref` довольно простая конструкция, мы можем сосредоточиться в основном на 
ее методах `get`, `set` и `of`, чтобы понять, как она работает.

### get и set подход

Допустим, у нас есть объект (назовем его `Shared`), который должен быть обновлен несколькими потоками. 
И мы используем для этого методы `get` и `set`, создав служебный метод, который поможет нам на этом пути:

```dotty
def modifyShared(trace: Ref[IO, Shared], msg: String): IO[Unit] =
  for
    sh <- trace.get()
    _ <- trace.set(Shared(sh, msg))
  yield ()
```

Наш объект `Shared` можно создать, 
применив для создания нового экземпляра его предыдущее состояние и новое значение.
`Shared` на самом деле может быть всем, чем мы захотим: простым списком, ассоциативным массивом 
или чем-то еще, к чему мы хотим получить одновременный безопасный доступ. 
Я только что изобрел `Shared(prev: Shared, msg: String)` для целей этого урока.

`F` в примере выше был заменен на конкретный `IO` из Cats effect, 
но имейте в виду, что `Ref` полиморфен на `F` и может использоваться с другими библиотеками.

С нашей монадической `IO` мы выполняем `flatMap` на каждом шаге 
и устанавливаем значение, хранящееся в нашей `Ref`, на желаемое — 
или... подождите, может быть, мы этого не делаем.

При таком подходе при одновременном вызове `modifyShared` мы можем потерять обновления! 
Это потому, что у нас может быть ситуация, когда, например, 
два потока могут прочитать значение с помощью `get`, и каждый из них будет выполнять `set` одновременно. 
Методы `get` и `set` не вызываются атомарно вместе.

### Атомарный update

Конечно, мы можем улучшить приведенный выше пример и использовать другие доступные методы из `Ref`. 
Для выполнения `get` и `set` вместе можно использовать `update`.

```dotty
def update(f: A => A): F[Unit]
```

Это решит проблему с обновлением значения, но у простого `update` есть свои недостатки. 
Если мы хотим прочитать значение сразу после его обновления, аналогично тому, как мы использовали `get` и `set`, 
то мы можем в конечном итоге прочитать устаревшие данные, скажем, наш `Ref` будет содержать ссылку на простой `Int`:

```dotty
for
  _ <- someRef.update(_ + 1)
  curr <- someRef.get
  _ <- IO { println(s"current value is $curr") }
yield ()
```

### `modify` в помощь

Мы можем немного это улучшить, используя `modify`, который будет делать то же самое, что и `update`, 
но вернет обновленное значение для дальнейшего использования.

```dotty
def modify[B](f: A => (A, B)): F[B] =
  @tailrec
  def spin: B =
    val c = ar.get
    val (u, b) = f(c)
    if !ar.compareAndSet(c, u) then spin
    else b

  F.delay(spin)
```

Как видите, это почти точно такая же реализация, 
как в примере `AtomicInteger.incrementAndGet`, показанном вначале, только на Scala. 
Можно заметить, что `Ref` основан на `AtomicReference`, чтобы делать свою работу.

### Ограничения Ref

Вы, наверное, уже заметили, что в случае, если значение не удастся обновить, 
то функция, переданная в `update`/`modify`, 
должна запускаться недетерминировано и, возможно, запускаться несколько раз. 
Хорошая новость заключается в том, что это решение в целом оказывается намного быстрее, 
чем стандартный механизм блокировки и синхронизации, 
и намного безопаснее, поскольку это решение не может блокироваться.

Как только мы узнали, как просто работает `Ref`, то можем перейти к другому Cats Concurrent классу: `Deferred`.

## Deferred

В отличие от `Ref`, `Deferred`:

- создан пустой
- можно выполнить только один раз
- и однажды установленный, он не может быть изменен или снова стать пустым.

Эти свойства делают `Deferred` простым и довольно интересным одновременно.

```dotty
abstract class Deferred[F[_], A]:
  def get: F[A]
  def complete(a: A): F[Unit]
```

`Deferred` предназначен для **чисто функциональной синхронизации**: 
когда мы вызываем `get` на пустом `Deferred`, мы блокируем его до тех пор, пока значение не станет доступно. 
[Согласно документации самого класса](https://typelevel.org/cats-effect/docs/std/deferred) блокировка:

> упомянутая блокировка является только семантической, никакие фактические потоки реализацией не блокируются

Тот же вызов `get` на непустом `Deferred` немедленно вернет сохраненное значение.

Другой метод - `complete` - установит значение, если экземпляр пуст, 
но при вызове на непустом `Deferred` вернет ошибку (failed IO в случае IO).

Здесь важно отметить, что `Deferred` требует, чтобы `F` был `Concurrent`, а это означает, что его можно отменить.

Хороший пример использования `Deferred` — это когда одна часть вашего приложения должна ждать другую. 
Пример ниже взят из отличного выступления Fabio Labella на Scala Italy 2019 — 
[Composable Concurrency with Ref + Deferred](https://vimeo.com/366191463), доступного на Vimeo.

```dotty
def consumer(done: Deferred[IO, Unit]) =
  for
    c <- Consumer.setup
    _ <- done.complete(())
    msg <- c.read
    _ <- IO(println(s"Received $msg"))
  yield ()

def producer(done: Deferred[IO, Unit]) =
  for
    p <- Producer.setup()
    _ <- done.get
    msg = "Msg A"
    _ <- p.write(msg)
    _ <- IO(println(s"Sent $msg"))
  yield ()

def prog =
  for
    d <- Deferred[IO, Unit]
    _ <- consumer(d).start
    _ <- producer(d).start
  yield ()
```

В приведенном выше примере у нас есть производитель и потребитель, 
и мы хотим, чтобы производитель дождался завершения настройки потребителя, прежде чем начать писать сообщения, 
иначе все, что мы напишем в производителе, будет потеряно. 
Чтобы преодолеть эту проблему, мы можем использовать общий экземпляр `Deferred` 
и блокировать его `get` до тех пор, пока `Deferred` экземпляр `done` не будет заполнен значением 
на стороне потребителя (в данном случае простое `Unit` значение `()`).

Конечно, приведенное выше решение не лишено проблем: 
если установка потребителя никогда не завершается, мы застреваем в ожидании, 
а производитель не может отправлять какие-либо сообщения. 
Чтобы преодолеть это, мы можем использовать тайм-аут на `get`, 
а также использовать `Either[Throwable, Unit]` 
или какую-либо другую конструкцию вместо простого `Unit` внутри объекта `Deferred`.

`Deferred` довольно прост, но в сочетании с `Ref` их можно использовать для 
создания более сложных структур данных, таких как, например, семафоры.

Для получения дополнительной информации рекомендуется посетить 
[саму документацию Cats](https://typelevel.org/cats-effect/docs/getting-started), 
где вы можете узнать больше о параллелизме в Cats и структурах данных, которые он предоставляет.

---

**Ссылки:**

- [Оригинал статьи - Cats Concurrency basics with Ref and Deferred][Ref and Deffered]

[Ref and Deffered]: https://blog.softwaremill.com/cats-concurrency-basics-with-ref-and-deferred-399e0335630
